package com.atguigu.edu.realtime.app.func;


import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.EduConfig;
import com.atguigu.edu.realtime.utils.HbaseUtil;
import com.atguigu.edu.realtime.utils.RedisUtil;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.util.Collections;


//维度关联模板模式
public abstract class MyAsyncFunction<T> extends RichAsyncFunction<T,T> implements JoinFunction<T> {
    private StatefulRedisConnection<String, String> RedisConnection;
    private AsyncConnection HbaseConnection;
    private String tableName;

    public MyAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        RedisConnection = RedisUtil.getAsyncRedisConnection();
        HbaseConnection = HbaseUtil.getAsyncHbaseConnection();
    }

    @Override
    public void close() throws Exception {
        RedisUtil.closeAsyncRedisConnection(RedisConnection);
        HbaseUtil.closeAsyncHbaseConnection(HbaseConnection);
    }

    @Override
    public void asyncInvoke(T orderBean, ResultFuture<T> resultFuture) throws Exception {

        //获取rowKey
        String key = getKey(orderBean);
        String rowKey = tableName + ":" + key;
        //从redis判断有没有
        JSONObject jsonObject = RedisUtil.asyncGetDimInfo(RedisConnection, rowKey);
        //如果redis没有,从hbase读取,并将结果写入到redis中
        if (jsonObject == null ){
            System.out.println("读取hbase--------------");
            jsonObject = HbaseUtil.getDimInfoFromHbaseByAsync(HbaseConnection, EduConfig.HBASE_NAMESPACE, tableName, key);
            if (jsonObject != null){
                RedisUtil.asyncWriteDim(RedisConnection,rowKey,jsonObject);
            }
        } else {
            System.out.println("读取redis-------------");
        }
        //将读取的维度字段封装到对象中
        if (jsonObject!=null){
            join(jsonObject,orderBean);
            resultFuture.complete(Collections.singleton(orderBean));
        }

    }

}
